package service

import bean.{OrdersToMysql, Store}
import dao.BatchDataDao
import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.functions.{avg, expr, hour, lit, to_timestamp}
import org.apache.spark.sql.types.FloatType
import org.apache.spark.sql.{Dataset, SaveMode, functions}
import util.{JBDCUtil, SparkUtil}



/**
 * @作者 YanTianCheng
 * @时间 2024/2/15 22:23
 * @文件: BatchDataService
 * @项目 离线分析
 */
class BatchDataService extends Serializable{
  val spark = SparkUtil.takeSpark()
  import spark.implicits._
  val dbProperties =  JBDCUtil.getDbProperties()
  val url = dbProperties.getProperty("url")
  val Properties = dbProperties.getProperty("user")
  dbProperties.getProperty("password")
  dbProperties.getProperty("driver")

  def dataAnalysis(): Unit = {
    //离线分析，对历史数据进行分析历史数据一般会存在数据库中（MySQL/HBase）
    //1.连接数据库
    //应该放在dao层
    val takeawayDao = new BatchDataDao
    val OrdersInfoDS = takeawayDao.getTakeawayData()
    val StoreInfoDS = takeawayDao.getTakeawayData2()
    val UserInfoDS = takeawayDao.getTakeawayData3()
    //需求一：离线统计热门城市Top10
//    hotCityCountTop10(OrdersInfoDS,StoreInfoDS)
    //需求二：离线统计高评分餐厅Top10
//    highRatingRestaurantTop10(OrdersInfoDS,StoreInfoDS)
    //需求三：离线统计热门菜品Top3
//    hotFoodCategoryTop3(OrdersInfoDS,StoreInfoDS)
    //需求四：离线统计高峰时间段，将时间段划分为早上、中午、下午和晚上
//    peakTime(OrdersInfoDS)
    //需求五：离线统计用户下单的平均时间
//    averageOrderTime(OrdersInfoDS)
    //需求六：起送价格和销量关系
    startUpPriceAndSales(StoreInfoDS)
  }

  //需求一：离线统计热门城市Top10
  def hotCityCountTop10(OrdersInfoDS: Dataset[OrdersToMysql], StoreInfoDS: Dataset[Store]): Unit = {
    // 将OrdersInfoDS与StoreInfoDS通过餐饮店ID关联
    val joinedDS = OrdersInfoDS.as("orders")
      .join(StoreInfoDS.as("stores"), $"orders. store_id " === $"stores.store_id")
    // 根据城市分组并统计数量
    val cityCountDS = StoreInfoDS.groupBy($"province",$"city").count()
    // 按照数量倒序排序并取前10个
    val top10DS = cityCountDS.orderBy($"count".desc).limit(10)
    // 转换DataFrame结构以满足Batch_Hot_City_Top10表的字段要求
    val resultDF = top10DS.select($"province", $"city", $"count" as "total_count")
    resultDF.write.mode(SaveMode.Overwrite)
      .jdbc(url, "Batch_Hot_City_Top10", dbProperties)
    println("---------离线统计热门城市Top10---------")
    // 显示结果
    resultDF.show()
  }
  //需求二：离线统计高评分餐厅Top10
  def highRatingRestaurantTop10(OrdersInfoDS: Dataset[OrdersToMysql], StoreInfoDS: Dataset[Store]): Unit = {
    val joinedDS = OrdersInfoDS.as("orders")
      .join(StoreInfoDS.as("stores"), $"orders.store_id" === $"stores.store_id")
    val restaurantAvgScoreDF = joinedDS.groupBy($"stores.store_id",$"stores.store_name").agg(avg($"orders.orders_score").as("avg_score"))
    val restaurantAvgScoreTop10DF = restaurantAvgScoreDF.orderBy($"avg_score".desc).limit(10)
    val resultDF = restaurantAvgScoreTop10DF.select($"stores.store_id", $"stores.store_name", $"avg_score")
    resultDF.write.mode(SaveMode.Overwrite)
      .jdbc(url, "Batch_High_Rating_Restaurant_Top10", dbProperties)
    println("---------离线统计高评分餐厅Top10---------")
    // 显示结果
    resultDF.show()
  }

  //需求三：离线统计热门菜品Top3
  def hotFoodCategoryTop3(OrdersInfoDS: Dataset[OrdersToMysql], StoreInfoDS: Dataset[Store]): Unit = {
    val joinedDS = OrdersInfoDS.as("orders")
      .join(StoreInfoDS.as("stores"), $"orders.store_id" === $"stores.store_id")
    val foodCategoryCountDS = joinedDS.groupBy($"stores.category").count()
    val foodCategoryCountTop3DS = foodCategoryCountDS.orderBy($"count".desc).limit(3)
    val resultDF = foodCategoryCountTop3DS.select($"category", $"count" as "total_count")
    resultDF.write.mode(SaveMode.Overwrite)
      .jdbc(url, "Batch_Hot_Food_Category_Top3", dbProperties)
    println("---------离线统计热门菜品Top3---------")
    // 显示结果
    resultDF.show()
  }

  //需求四：离线统计高峰时间段，将时间段划分为早上、中午、下午和晚上
  def peakTime(OrdersInfoDS: Dataset[OrdersToMysql]): Unit = {
    // 假设OrdersToMysql是一个定义了case class OrdersToMysql的Dataset
    // 并且它包含create_time和delivery_time字段，它们都是字符串类型

    // 将创建时间转换为Timestamp类型
    val OrdersDSWithTime = OrdersInfoDS
      .withColumn("createTime", to_timestamp($"create_time", "yyyy-MM-dd HH:mm:ss"))
      .withColumn("deliveryTime", to_timestamp($"delivery_time", "yyyy-MM-dd HH:mm:ss"))

    // 定义窗口规范
    val windowSpec = Window.partitionBy().orderBy($"createTime").rangeBetween(Window.unboundedPreceding, Window.currentRow)

    // 根据创建时间统计各个时间段的订单数
    val hourlyOrderCounts = OrdersDSWithTime
      .withColumn("hour", hour($"createTime")) // 添加小时列
      .groupBy($"hour")
      .agg(functions.count("*").as("orderCount"))

    // 添加时间段标签并统计
    val peakTimes = hourlyOrderCounts
      .withColumn("timePeriod", expr("CASE WHEN hour >= 6 AND hour < 12 THEN '早上' " +
        "WHEN hour >= 12 AND hour < 14 THEN '中午' " +
        "WHEN hour >= 14 AND hour < 18 THEN '下午' " +
        "ELSE '晚上' END")) // 使用expr和CASE WHEN来定义时间段
      .groupBy("timePeriod")
      .agg(functions.sum("orderCount").as("totalOrders"))
    peakTimes.write.mode(SaveMode.Overwrite)
      .jdbc(url, "Batch_Peak_Time", dbProperties)
    println("---------离线统计高峰时间段---------")
    peakTimes.show()
  }

  //需求五：离线统计订单配送时间的平均时间
  def averageOrderTime(OrdersInfoDS: Dataset[OrdersToMysql]): Unit = {
    // 将delivery_time的字符串类型转换为整数类型（分钟）
    val dsWithDeliveryTimeAsInt = OrdersInfoDS.withColumn("delivery_time", $"delivery_time".cast("integer"))

    // 按store_id分组并计算每个商家的平均delivery_time
    val avgDeliveryTimesByStore = dsWithDeliveryTimeAsInt.groupBy("store_id")
      .agg(avg("delivery_time").alias("avg_delivery_time"))
    avgDeliveryTimesByStore.write.mode(SaveMode.Overwrite)
      .jdbc(url, "Batch_Average_Order_Time", dbProperties)
    println("---------离线统计用户下单的平均时间---------")
    avgDeliveryTimesByStore.show()
  }


  //需求六：起送价和销量关系
  def startUpPriceAndSales(StoreInfoDS: Dataset[Store]): Unit = {
    // 将字符串类型的min_price列转换为浮点数类型以便进行计算
    val convertedDS = StoreInfoDS.withColumn("min_price", $"min_price".cast(FloatType))
    // 计算min_price的平均值
    val avgMinPrice = convertedDS.agg(avg("min_price")).first.getAs[Double]("avg(min_price)")
    // 从原始数据集中选择需要写入数据库的字段并添加平均价格列
    val resultDS = StoreInfoDS.select($"store_id", $"min_price", lit(avgMinPrice).as("avg_min_price"), $"month_num")
    resultDS.write.mode(SaveMode.Overwrite)
      .jdbc(url, "Batch_Start_Up_Price_And_Sales", dbProperties)
    println("---------起送价和销量关系---------")
    resultDS.show()
  }

}
